package com.yin.myproject;

import com.yin.myproject.core.IScheduleDataManager;
import com.yin.myproject.core.ScheduleServer;
import com.yin.myproject.zk.ScheduleDataManager4ZK;
import com.yin.myproject.zk.ZKManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 调度核心管理器
 *
 * @author Eason
 * @create 2018-04-08 10:36
 **/
public class ZKScheduleManager {
    private static final transient Logger LOGGER = LoggerFactory.getLogger(ZKScheduleManager.class);
    private Map<String, String> zkConfig;

    private ZKManager zkManager;

    private IScheduleDataManager scheduleDataManager;

    /**
     * 当前调度服务的信息
     */
    protected ScheduleServer currenScheduleServer;

    /**
     * 是否启动调度管理，如果只是做系统管理，应该设置为false
     */
    private boolean start = true;

    /**
     * 心跳间隔
     */
    private int timerInterval = 2000;

    /**
     * 是否注册成功
     */
    private boolean isScheduleServerRegister = false;

    private Map<String, Boolean> isOwnerMap = new ConcurrentHashMap<String, Boolean>();

    private Timer hearBeatTimer;
    private Lock initLock = new ReentrantLock();
    private boolean isStopSchedule = false;
    private Lock registerLock = new ReentrantLock();

    private volatile String errorMessage = "No config Zookeeper connect information";
    private InitialThread initialThread;

    public ZKScheduleManager() {
        this.currenScheduleServer = ScheduleServer.createScheduleServer(null);
    }

    public void init() throws Exception {
        Properties properties = new Properties();
        for (Map.Entry<String, String> e : this.zkConfig.entrySet()) {
            properties.put(e.getKey(), e.getValue());
        }
        this.init(properties);
    }

    /**
     * 重新初始化
     * @param p
     * @throws Exception
     */
    public void reInit(Properties p) throws Exception {
        if (this.start || this.hearBeatTimer != null) {
            throw new Exception("调度器有任务处理，不能重新初始化");
        }
        this.init(p);
    }

    public void init(Properties p) throws Exception {
        if (this.initialThread != null) {
            this.initialThread.stopThread();
        }
        this.initLock.lock();
        try {
            this.scheduleDataManager = null;
            if (this.zkManager != null) {
                this.zkManager.close();
            }
            this.zkManager = new ZKManager(p);
            this.errorMessage = "Zookeeper connecting ......" + this.zkManager.getConnectStr();
            initialThread = new InitialThread(this);
            initialThread.setName("ScheduleManager-initialThread");
            initialThread.start();
        } finally {
            this.initLock.unlock();
        }
    }

    private void rewriteScheduleInfo() throws Exception {
        registerLock.lock();
        try {
            if (this.isStopSchedule) {
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("外部命令终止调度,不再注册调度服务，避免遗留垃圾数据：" + currenScheduleServer.getUuid());
                }
                return;
            }
            // 先发送心跳信息
            if (errorMessage != null) {
                this.currenScheduleServer.setDealInfoDesc(errorMessage);
            }
            if (!this.scheduleDataManager.refreshScheduleServer(this.currenScheduleServer)) {
                // 更新信息失败，清除内存数据后重新注册
                this.clearMemoInfo();
                this.scheduleDataManager.registerScheduleServer(this.currenScheduleServer);
            }
            isScheduleServerRegister = true;
        } finally {
            registerLock.unlock();
        }
    }

    /**
     * 清除内存中所有的已经取得的数据和任务队列,在心态更新失败，或者发现注册中心的调度信息被删除
     */
    public void clearMemoInfo() {
        try {

        } finally {
        }

    }


    /**
     * 根据当前调度服务器的信息，重新计算分配所有的调度任务 任务的分配是需要加锁，避免数据分配错误。为了避免数据锁带来的负面作用，通过版本号来达到锁的目的<br/>
     * <ol>
     * <li>获取任务状态的版本号</li>
     * <li>获取所有的服务器注册信息和任务队列信息</li>
     * <li>清除已经超过心跳周期的服务器注册信息</li>
     * <li>重新计算任务分配</li>
     * <li>更新任务状态的版本号【乐观锁】</li>
     * <li>根系任务队列的分配信息</li>
     * </ol>
     *
     * @throws Exception
     */
    public void assignScheduleTask() throws Exception {
        scheduleDataManager.clearExpireScheduleServer();
        List<String> serverList = scheduleDataManager.loadScheduleServerNames();
        if (!scheduleDataManager.isLeader(this.currenScheduleServer.getUuid(), serverList)) {
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug(this.currenScheduleServer.getUuid() + ":不是负责任务分配的Leader,直接返回");
            }
            return;
        }
        // 黑名单
        for (String ip : zkManager.getIpBlacklist()) {
            int index = serverList.indexOf(ip);
            if (index > -1) {
                serverList.remove(index);
            }
        }
        // 设置初始化成功标准，避免在leader转换的时候，新增的线程组初始化失败
        scheduleDataManager.assignTask(this.currenScheduleServer.getUuid(), serverList);
    }

    /**
     * 定时向数据配置中心更新当前服务器的心跳信息。 如果发现本次更新的时间如果已经超过了，服务器死亡的心跳周期，则不能在向服务器更新信息。 而应该当作新的服务器，进行重新注册。
     *
     * @throws Exception
     */
    public void refreshScheduleServer() throws Exception {
        try {
            rewriteScheduleInfo();
            // 如果任务信息没有初始化成功，不做任务相关的处理
            if (!this.isScheduleServerRegister) {
                return;
            }

            // 重新分配任务
            this.assignScheduleTask();
            // 挑拣当前调度服务的任务
            this.checkLocalTask();
        } catch (Exception e) {
            // 清除内存中所有的已经取得的数据和任务队列,避免心跳线程失败时候导致的数据重复
            this.clearMemoInfo();
            throw e;
        }
    }

    public void checkLocalTask() throws Exception {
        // 检查系统任务执行情况
        LOGGER.debug("schedule the currentServer[{}] tasks.", this.currenScheduleServer.getUuid());
        scheduleDataManager.checkLocalTask(this.currenScheduleServer.getUuid());
    }

    /**
     * 在Zk状态正常后回调数据初始化
     *
     * @throws Exception
     */
    public void initialData() throws Exception {

        LOGGER.info("init ScheduleDataManager ...");

        this.zkManager.initial();
        this.scheduleDataManager = new ScheduleDataManager4ZK(this.zkManager);
        if (this.start) {
            // 注册调度管理器
            this.scheduleDataManager.registerScheduleServer(this.currenScheduleServer);
            if (hearBeatTimer == null) {
                hearBeatTimer = new Timer("ScheduleManager-" + this.currenScheduleServer.getUuid() + "-HearBeat");
            }
            hearBeatTimer.schedule(new HeartBeatTimerTask(this), 2000, this.timerInterval);
        }
        LOGGER.info("init ScheduleDataManager finished.");
    }

    private Runnable taskWrapper(final Runnable task) {
//        return new Runnable() {
//            public void run() {
//                String scheduleKey = null;
//                if (task instanceof ScheduledMethodRunnable) {
//                    ScheduledMethodRunnable backendScheduledMethodRunnable = (ScheduledMethodRunnable) task;
//                    scheduleKey = backendScheduledMethodRunnable.getKey();
//                } else {
//                    org.springframework.scheduling.support.ScheduledMethodRunnable springScheduledMethodRunnable = (org.springframework.scheduling.support.ScheduledMethodRunnable) task;
//                    String[] beanNames = ApplicationContextUtil.getContext().getBeanNamesForType(
//                            springScheduledMethodRunnable.getTarget().getClass());
//                    if (beanNames != null) {
//                        scheduleKey = beanNames[0];
//                    }
//                }
//                if (!Strings.isNullOrEmpty(scheduleKey)) {
//                    boolean isOwner = false;
//                    try {
//                        if (!isScheduleServerRegister) {
//                            Thread.sleep(1000);
//                        }
//                        if (zkManager.checkZookeeperState()) {
//                            isOwner = scheduleDataManager.isOwner(scheduleKey, currenScheduleServer.getUuid());
//                            isOwnerMap.put(scheduleKey, isOwner);
//                        } else {
//                            // 如果zk不可用，使用历史数据
//                            if (null != isOwnerMap) {
//                                isOwner = isOwnerMap.get(scheduleKey);
//                            }
//                        }
//                        if (isOwner) {
//                            task.run();
//                            scheduleDataManager.saveRunningInfo(scheduleKey, currenScheduleServer.getUuid());
//                            LOGGER.info("Job[taskKey: {}] has been executed at server: {}.", scheduleKey,
//                                    currenScheduleServer.getUuid());
//                        }
//                    } catch (Exception e) {
//                        LOGGER.error("Check task owner error.", e);
//                    }
//                }
//            }
//        };
        return null;
    }

    class HeartBeatTimerTask extends java.util.TimerTask {
        private transient final Logger log = LoggerFactory.getLogger(HeartBeatTimerTask.class);
        ZKScheduleManager manager;

        public HeartBeatTimerTask(ZKScheduleManager aManager) {
            manager = aManager;
        }

        public void run() {
            try {
                Thread.currentThread().setPriority(Thread.MAX_PRIORITY);
                manager.refreshScheduleServer();
            } catch (Exception ex) {
                log.error(ex.getMessage(), ex);
            }
        }
    }

    class InitialThread extends Thread {
        private transient Logger log = LoggerFactory.getLogger(InitialThread.class);
        ZKScheduleManager sm;

        public InitialThread(ZKScheduleManager sm) {
            this.sm = sm;
        }

        boolean isStop = false;

        public void stopThread() {
            this.isStop = true;
        }

        @Override
        public void run() {

            log.info("run init thread.");

            sm.initLock.lock();
            try {
                int count = 0;
                while (!sm.zkManager.checkZookeeperState()) {
                    count = count + 1;
                    if (count % 50 == 0) {
                        sm.errorMessage = "Zookeeper connecting ......" + sm.zkManager.getConnectStr() + " spendTime:"
                                + count * 20 + "(ms)";
                        log.error(sm.errorMessage);
                    }
                    Thread.sleep(20);
                    if (this.isStop) {
                        return;
                    }
                }
                sm.initialData();
                log.info("init ZKScheduleManager finished.");
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            } finally {
                sm.initLock.unlock();
            }

        }

    }

    public IScheduleDataManager getScheduleDataManager() {
        return scheduleDataManager;
    }

    public void setZkManager(ZKManager zkManager) {
        this.zkManager = zkManager;
    }

    public ZKManager getZkManager() {
        return zkManager;
    }

    public void setZkConfig(Map<String, String> zkConfig) {
        this.zkConfig = zkConfig;
    }

    public String getScheduleServerUUid() {
        if (null != currenScheduleServer) {
            return currenScheduleServer.getUuid();
        }
        return null;
    }

    public Map<String, Boolean> getIsOwnerMap() {
        return isOwnerMap;
    }

//    @Override
//    public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, long period) {
//        return super.scheduleAtFixedRate(taskWrapper(task), period);
//    }
//
//    @Override
//    public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {
//        return super.schedule(taskWrapper(task), trigger);
//    }
//
//    @Override
//    public ScheduledFuture<?> schedule(Runnable task, Date startTime) {
//        return super.schedule(taskWrapper(task), startTime);
//    }
//
//    @Override
//    public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Date startTime, long period) {
//        return super.scheduleAtFixedRate(taskWrapper(task), startTime, period);
//    }
//
//    @Override
//    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Date startTime, long delay) {
//        return super.scheduleWithFixedDelay(taskWrapper(task), startTime, delay);
//    }
//
//    @Override
//    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, long delay) {
//        return super.scheduleWithFixedDelay(taskWrapper(task), delay);
//    }

    public boolean isReady() {
        return this.isScheduleServerRegister && (this.scheduleDataManager != null) && (this.zkManager != null);
    }
}
